package com.uuuo.rxjava1;

import org.junit.Test;

import java.util.ArrayList;
import java.util.List;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.observables.GroupedObservable;

/**
 * Created by JiaTao on 2017/4/1.
 */

public class B_Transfer extends A_Base {
    @Test
    public void flatMap() {
        class Course {
            private String name;

            public Course(String name) {
                this.name = name;
            }

            public String getName() {
                return name;
            }
        }
        class Student {
            private List<Course> courses;

            public Student(List<Course> courses) {
                this.courses = courses;
            }

            public List<Course> getCourses() {
                return courses;
            }
        }

        /* Student1 */
        Course course11 = new Course("C#");
        Course course12 = new Course("Java");
        Course course13 = new Course("Python");
        List<Course> courses1 = new ArrayList<Course>();
        courses1.add(course11);
        courses1.add(course12);
        courses1.add(course13);
        Student student1 = new Student(courses1);

        /* Student2 */
        Course course21 = new Course("OC");
        Course course22 = new Course("Ruby");
        List<Course> courses2 = new ArrayList<Course>();
        courses2.add(course21);
        courses2.add(course22);
        Student student2 = new Student(courses2);

        Observable.just(student1, student2)
                .flatMap(new Func1<Student, Observable<Course>>() {
                    @Override
                    public Observable<Course> call(Student student) {
                        return Observable.from(student.getCourses());
                    }
                })
                .toList()
                .subscribe(new Action1<List<Course>>() {
                    @Override
                    public void call(List<Course> courses) {
                        showMsg(courses.size()+"");
                    }
                });
//                .subscribe(new Subscriber<Course>() {
//                    @Override
//
//                    public void onCompleted() {
//                        System.out.println("onCompleted.");
//                    }
//
//                    @Override
//                    public void onError(Throwable e) {
//                        System.out.println("onError: " + e.getMessage());
//                    }
//
//                    @Override
//                    public void onNext(Course course) {
//                        System.out.println("onNext: " + course.getName());
//                    }
//                });
    }

    @Test
    public void scan() {
//        Scan操作符对原始Observable发射的第一项数据应用一个函数，然后将那个函数的结果作为自己的第一项数据发射。
//        它将函数的结果同第二项数据一起填充给这个函数来产生它自己的第二项数据。
//        它持续进行这个过程来产生剩余的数据序列。
//        这个操作符在某些情况下被叫做accumulator。
        Observable
                .just(0, 1, 2, 3, 4, 5, 6)
                .scan(new Func2<Integer, Integer, Integer>() {
                    @Override
                    public Integer call(Integer integer, Integer integer2) {
                        return integer + 1;
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        showMsg(integer + "");
                    }
                });
    }

    @Test
    public void groupBy() {
// 完整流程版
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
                .groupBy(new Func1<Integer, Integer>() {
                    @Override
                    public Integer call(Integer integer) {
                        showMsg("groupBy---->" + integer + "groupBy%3---->" + integer % 3);
                        return integer % 3;
                    }
                })
                .subscribe(new Subscriber<GroupedObservable<Integer, Integer>>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(final GroupedObservable<Integer, Integer> integerIntegerGroupedObservable) {
                        integerIntegerGroupedObservable.subscribe(new Subscriber<Integer>() {
                            @Override
                            public void onCompleted() {

                            }

                            @Override
                            public void onError(Throwable e) {

                            }

                            @Override
                            public void onNext(Integer integer) {
                                showMsg(integerIntegerGroupedObservable.getKey() + "----->" + integer);
                            }
                        });
                    }
                });
    }

    @Test
    public void buffer() {
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
                .buffer(3, 3)
                .subscribe(new Subscriber<List<Integer>>() {
                    @Override
                    public void onCompleted() {
                        System.out.println("onCompleted.");
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(List<Integer> integers) {
                        System.out.println("onNext: " + integers);
                    }
                });
    }

    @Test
    public void window() {
// 简化版
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
                .window(3)
                .subscribe(new Action1<Observable<Integer>>() {
                    //                    window操作符与buffer操作符类似，区别在于buffer操作符产生的结果是一个List缓存，
//                    而window操作符产生的结果是一个Observable,
//                    订阅者可以对这个结果Observable重新进行订阅处理。
                    @Override
                    public void call(Observable<Integer> integerObservable) {
                        System.out.println("start");
                        integerObservable.subscribe(new Action1<Integer>() {
                            @Override
                            public void call(Integer integer) {
                                System.out.println("onNext: " + integer);
                            }
                        });
                    }
                });
    }
}
